package com.atguigu.edu.app.dws;

import app.func.KeywordUDTF;
import bean.KeywordBean;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import util.MyClickhouseUtil;
import util.MyKafkaUtil;

/**
 * 使用SQL的方式实现关键词统计
 */
public class DwsTrafficKeywordWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //注册自定义的UDTF函数
        tableEnv.createTemporarySystemFunction("ik_analyze", KeywordUDTF.class);
        //TODO 2.检查点相关设置

        //TODO 3.从kafka的page_log主题中读取数据，创建动态表，指定Watermark以及事件时间字段
        String topic = "dwd_traffic_page_log";
        String groupId= "dws_traffic_keyword_group";
        tableEnv.executeSql("CREATE TABLE page_log (\n" +
                "    common map<string,string>,\n" +
                "    page map<string,string>,\n" +
                "    ts BIGINT,\n" +
                "    row_time as TO_TIMESTAMP(FROM_UNIXTIME(ts/1000)) ,\n" +
                "    WATERMARK FOR row_time AS row_time - INTERVAL '3' SECOND\n" +
                ")" + MyKafkaUtil.getKafkaDDL(topic,groupId));

        //TODO 4.过滤出关键词搜索数据
        Table searchTable = tableEnv.sqlQuery("select page['item'] full_word,row_time " +
                "from page_log where page['item_type']='keyword' and page['item']  is not null");
        tableEnv.createTemporaryView("search_table",searchTable);
        //TODO 5.使用自定义UDTF函数 对搜索内容进行分词 并将分词的结果和表中的字段连接
        Table splitTable = tableEnv.sqlQuery("select keyword,row_time from search_table, lateral table(ik_analyze(full_word)) t(keyword)");
        tableEnv.createTemporaryView("split_table",splitTable);

        //TODO 6.分组、开窗、聚合计算
        Table resTable = tableEnv.sqlQuery("select " +
                " DATE_FORMAT(TUMBLE_START(row_time, INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') stt," +
                " DATE_FORMAT(TUMBLE_END(row_time, INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') edt," +
                " keyword," +
                " count(*) keyword_count," +
                " UNIX_TIMESTAMP()*1000 ts" +
                " from split_table group by TUMBLE(row_time, INTERVAL '10' SECOND),keyword");

        //TODO 7.将动态表转化为流(查询的字段名称和实体类对象属性名称保持一致)
        DataStream<KeywordBean> keywordDS = tableEnv.toAppendStream(resTable, KeywordBean.class);
        keywordDS.print(">>>");
        //TODO 8.将流中的结果写到clickhouse中
        keywordDS.addSink(MyClickhouseUtil.getSinkFunction("insert into dws_traffic_keyword_page_view_window values(?,?,?,?,?)"));
        env.execute();
    }
}
